Nit: just call it IndexedFile? Kernel is just an impl details

+1

Done.

let's follow https://github.com/delta-io/delta/blob/master/kernel/kernel-api/src/main/java/io/delta/kernel/internal/actions/RowBackedAction.java#L46 and create a new helper function here

protected int getFieldIndex(String fieldName) {
    int index = row.getSchema().indexOf(fieldName);
    checkArgument(index >= 0, "Field '%s' not found in schema: %s", fieldName, row.getSchema());
    return index;
  }

Done. Thank you!

I wonder if we can following https://github.com/delta-io/delta/blob/master/kernel-spark/src/test/java/io/delta/kernel/spark/read/SparkGoldenTableTest.java#L579 and test many of the golden tables.

I think it might be overkill at this point, especially when there are so many unsupported table types. I agree we should eventually add a test like this. Added a TODO.

Should validate happen after processing previous version?

Done, thanks!

Why do you need to get a snapshot even if we start reading from a specific delta log version?

It's required by the kernel to fetch actions:

Ideally this is class static variable so it will only be allocated once per query run.

Why do we also need to get the "REMOVE" actions?

Done.

See validateCommit() -- the current behavior of the delta connector is that we fail the pipeline if any commit contains a REMOVE (unless skipDeletes or skipChangeCommits are specified). Streaming jobs are meant to process append-only data

Where are you filtering out the "REMOVE" actions?

We throw an error whenever we encounter a REMOVE -- because ETL jobs should process append-only data. We fail explicitly to avoid correctness issues.
In M2, we'll also support ignoreChangedCommits and ignoreDeletes to skip these commits silently.
To properly handle update actions, users would need to use CDF.

Should we throw here? given -- addVector.isNullAt(rowId) is false

We call this method even on REMOVE rows in extractIndexedFilesFromBatch

if this is a remove row, iiuc the method will return on L175? Did I miss something?

Ah yes. You are right. Done.

Its more performant to use a linked list if we don't actually know the size the list will be

I don't think so. The per-node overhead of a linked list outweighs the cost of resizing of an arraylist, especially for my use case (addAll(), add(), clear()). We would maybe get a performance benefit if we do a lot of deletes and inserts at the beginning or in the middle (which we are not).

This is not very efficient. "dataFiles" should just be a linked list and you can append and prepend in constant time.

ditto -- I don't think linked lists would help here.

This logic here is kind of confusing. All you are trying to do is sandwich the index files between the BASE_INDEX sentinel file and END_INDEX sentinel file right? Why not simplify the logic to be

allIndexedFiles.add(beginSentinelFile)
allIndexedFiles.addAll(allIndexFilesInBatch)
allIndexedFiles.add(endSentinelFile)

We only insert sentinels before and after a version. The code is complex because the kernel breaks up a commit into batches (ColumnarBatch) to avoid overwhelming memory. I reorganized the code a bit to make this clear. Could you take another look?

Chatted offline with @zikangh. I had a concern on why we needed to use another list i.e. "currentVersionFiles" to buffer files for this version and not just directly append to allIndexedFiles. The reason is that in the next PR, she is going to introduce the ability to skip whole commits.

Use linkedlist

Same rationale as above -- we are only doing addAll() and add(), arrayList would be faster and more memory-efficient.

What about to and from END_INDEX?

Done. Thanks!

Should we also test with other types of actions in the delta log?

We do test REMOVEs & ADDs, I added METADATA too (which now will yield empty commits).

nit: document that deltaFile.add() != null could happen when it is starting/ending index

Done.

is this duplicating any code that already present in v1 delta connector?

Yes we are duplicating the scala version of IndexedFile, because the logic is simple enough and pulls in DeltaLog-only dependencies AddFile and RemoveFile. If I create a common interface for AddFile and RemoveFile to bridge kernel and DeltaLog, the code would be harder-to-maintain and more error-prone.

However there are cases where refactoring is a clear win, e.g. DeltaSourceAdmissionBase and AdmissionLimits -- the logic is complex and a small refactor would make them work for both Kernel and DeltaLog classes.

w.r.t duplicating vs sharing code, I'm weighing the pros and cons on a case by case basis following the principles I outlined above.